package com.iussoft.gateway.filter;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.apache.tools.DateUtils;
import com.apache.tools.NumberUtils;
import com.apache.tools.StrUtil;
import com.apache.tools.UUIDUtil;
import com.google.common.util.concurrent.RateLimiter;
import com.iussoft.gateway.protocol.RequestProtocol;
import com.iussoft.gateway.protocol.impl.FtpGateWayPluginImpl;
import com.iussoft.gateway.protocol.impl.SoapGatewayPluginImpl;
import com.iussoft.tools.LoggerUtil;
import io.netty.buffer.UnpooledByteBufAllocator;
import org.reactivestreams.Publisher;
import org.springframework.cloud.gateway.filter.GatewayFilterChain;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.NettyDataBufferFactory;
import org.springframework.http.HttpStatus;
import org.springframework.http.server.reactive.ServerHttpResponse;
import org.springframework.stereotype.Component;
import org.springframework.web.server.ServerWebExchange;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.io.UnsupportedEncodingException;
import java.net.InetSocketAddress;
import java.net.URI;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.atomic.LongAdder;

import static org.springframework.cloud.gateway.support.ServerWebExchangeUtils.*;

/**
 * description:全局过滤器
 *
 * @author dyh 创建时间：2020年02月24日
 */
@Component
public class AuthSignatureFilter extends SuperGlobalFilter {

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        long startTime = System.currentTimeMillis();
        JSONObject route = getGatewayRoute(exchange);
        if(null == route || route.isEmpty()) {//判断路由是否存在
            Response json = new Response("404", "请求服务不存在", "请求服务不存在");
            return writeWith(exchange,json,startTime,false);
        }
        if("1".equals(route.getString("ifOpenHystrix"))){//开启熔断
            String uri =exchange.getRequest().getPath().value();
            long dd = Double.valueOf(StrUtil.doNull(route.getString("hstxCondition"),"1000")).longValue();
            String date = DateUtils.Now.fmt_yyyyMMdd();
            if (!map.containsKey(uri)) {
                Map<String,LongAdder> limit = new HashMap();
                limit.put(date,new LongAdder());
                map.put(uri,limit);
            }
            Map<String,LongAdder> limit = map.get(uri);
            boolean tryAcquire = true;
            if(limit.containsKey(date)){
                LongAdder lad = limit.get(date);
                if (dd >= lad.sum()) {
                    lad.increment();
                    tryAcquire = false;
                }
            } else {
                limit.clear();
                limit.put(date,new LongAdder());
                tryAcquire = false;
            }
            if (tryAcquire) {//达到熔断条件
                boolean isOpenLog = "1".equals(route.getString("ifOpenLog"));//是否开启打印日志
                Response json = new Response("200", "请求成功", "系统对此功能已暂停服务");
                return writeWith(exchange,json,startTime,isOpenLog);
            }
        }
        URI requestUrl = exchange.getRequest().getURI();
        final String scheme = requestUrl.getScheme();
        if (isAlreadyRouted(exchange) || "http".equals(scheme) || "https".equals(scheme) || "lb".equals(scheme) || "ws".equals(scheme) || "wss".equals(scheme)) {
            return globalProtocol(exchange,chain,route,startTime);
        }
        LoggerUtil.infoSource("请求的url为{}; 协议为{}",exchange.getRequest().getURI().toString(),scheme);
        setAlreadyRouted(exchange);
        RecorderServerHttpRequestDecorator requestDecorator = new RecorderServerHttpRequestDecorator(exchange.getRequest());
        AtomicReference<String> requestBody = requestDecorator.getRequestBody();
        /**
         * 获取请求的url 对路径进行重新编码
         */
        final String url = requestUrl.toASCIIString();
        return chain.filter(exchange).then(Mono.defer(() -> {
                    ServerHttpResponse response = exchange.getResponse();
                    return response.writeWith(Flux.create(sink -> {
                        NettyDataBufferFactory nettyDataBufferFactory = new NettyDataBufferFactory(new UnpooledByteBufAllocator(false));
                        JSONObject json = new JSONObject();
                        DataBuffer dataBuffer = null;
                        String charset = "UTF-8";
                        try {
                            RequestProtocol protocol = null;
                            if("ftp".equalsIgnoreCase(scheme)){
                                protocol = new FtpGateWayPluginImpl();
                            } else if("soap".equalsIgnoreCase(scheme)){
                                protocol = new SoapGatewayPluginImpl();
                            }
                            if(null != protocol) {
                                Object result = protocol.invoke(url, requestBody.get());
                                HttpStatus status = HttpStatus.resolve(500);
                                /**
                                 * 判断结果是否返回，如果没有数据则直接返回
                                 */
                                if (result != null) {
                                    json = (JSONObject) result;
                                    status = HttpStatus.resolve(json.getInteger("code"));
                                    json.remove("code");
                                }
                                response.setStatusCode(status);
                                try {
                                    dataBuffer = nettyDataBufferFactory.wrap(json.toJSONString().getBytes(charset));
                                } catch (UnsupportedEncodingException e) {
                                    String resStr = JSON.toJSONString(new Response("500", "请求失败", e.toString()));
                                    dataBuffer = nettyDataBufferFactory.wrap(resStr.getBytes(charset));
                                    LoggerUtil.error("返回调用请求数据错误:", e);
                                }
                            }
                        } catch (Exception e) {
                            try {
                                String resStr = JSON.toJSONString(new Response("500","请求失败",e.toString()));
                                dataBuffer = nettyDataBufferFactory.wrap(resStr.getBytes(charset));
                                LoggerUtil.error("获取远程数据错误:",e);
                            } catch (UnsupportedEncodingException ex) {
                                LoggerUtil.error("返回调用请求数据错误:",ex);
                            }
                        }
                        /**
                         * 将数据进行发射到下一个过滤器
                         */
                        sink.next(dataBuffer);
                        sink.complete();
                    }));
                }));
    }

    private Mono<Void> writeWith(ServerWebExchange exchange, Response json, long startTime,boolean isOpenLog){
        ServerHttpResponse response = exchange.getResponse();
        DataBuffer bodyDataBuffer = response.bufferFactory().wrap(JSON.toJSONString(json).getBytes());
        if(isOpenLog) {
            StringBuilder sb = new StringBuilder();
            String traceId = UUIDUtil.getPrimaryKey();
            sb.append("traceId = " + traceId);
            sb.append("; infoType = warning; requestUrl = " + exchange.getRequest().getURI());
            InetSocketAddress address = exchange.getRequest().getRemoteAddress();
            sb.append("; requestIp = " + address.getHostName() + ":" + address.getPort());
            sb.append("; requestTime = " + DateUtils.Now.fmt_yyyyMMdd_HHmmss());
            sb.append("; total = " + (System.currentTimeMillis() - startTime));
            sb.append("; errorMessage = " + json.getData() + "; exeSys = gateway; level = ERROR");
            LoggerUtil.warn(sb.toString());
        }
        return response.writeWith(Mono.just(bodyDataBuffer));
    }
}
